package com.rengwuxian.coursecoroutines._4_flow

import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.channels.BufferOverflow
import kotlinx.coroutines.channels.Channel.Factory.CONFLATED
import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.buffer
import kotlinx.coroutines.flow.collectLatest
import kotlinx.coroutines.flow.conflate
import kotlinx.coroutines.flow.flow
import kotlinx.coroutines.flow.flowOn
import kotlinx.coroutines.flow.map
import kotlinx.coroutines.flow.mapLatest
import kotlinx.coroutines.launch
import kotlinx.coroutines.runBlocking
import kotlinx.coroutines.withContext
import kotlin.concurrent.thread
import kotlin.coroutines.EmptyCoroutineContext

/**
 * buffer()系列操作符
 * buffer() / conflate() / collectLatest() 、 flowOn()、channelFlow()
 *
 * buffer()：给Flow加上缓冲功能，底层是Channel实现的。
 *
 * 把生产流程和处理流程分开（切线程），还要具有缓冲功能，暂存上游的生产数据。
 * flowOn()，默认把缓冲功能打开了，如果需要配置缓冲，则用buffer()
 * 说白了，flowOn和buffer创建的是同一个东西，只不过flowOn是用来配置CoroutineContext的，而buffer是用来配置缓冲。
 * 效果都是切协程+缓冲功能。
 * 底层都是用ChannelFlow这个类来实现的。是同一个东西。
 *
 * flowOn().buffer()也是存在融合特性。
 * buffer().buffer()的融合策略：右边优先，如果右边第二个参数不是SUSPEND，则右边的buffer会完全覆盖左边的buffer。
 * 如果第二个参数是SUSPEND，或者不填，则会融合两个buffer的缓冲区尺寸，相加。
 * 如果两个都没有填尺寸大小，则使用默认64，如果有一个填了尺寸，就用填了的尺寸大小。
 *
 * conflate()：buffer()的变种，它实际上就是一个只缓冲最新的一条数据的buffer().
 * 和buffer(CONFLATED)写法是一样的。
 *
 * collectLatest():是两个操作符的合并应用，mapLatest(action).buffer(0).collect()
 * mapLatest(action)：它是一个可以随时被新数据取消当前数据的转换流程的、增强版的map().
 *                    隐含了一个条件，就是在当前数据转换过程中，下一个数据就可以进行生产了。
 *                    即：mapLatest(action)也是需要多协程的支持的。
 *       mapLatest{}的处理过程和缓冲是没有关系的。
 *       flow1.mapLatest { it }.buffer(0)，关闭缓冲区。数据到来后，不是因为缓冲之后被扔掉了，而是没机会转换出来，根本就没有进到缓冲里。
 *                                         效果就是下游没有处理完成之前，上游就不再转换新数据的mapLatest{}
 */
fun main() = runBlocking<Unit> {
  val scope = CoroutineScope(EmptyCoroutineContext)
  val start = System.currentTimeMillis()
  val flow1 = flow {
    for (i in 1..5) {
      emit(i)
      println("Emitted: $i - ${System.currentTimeMillis() - start}ms")
    }
  }.buffer(1)
    .flowOn(Dispatchers.IO)
    .buffer(2)
//    .conflate()
    .map { it + 1 }
    .map { it * 2 }
  scope.launch {
    flow1.mapLatest { it }.buffer(0).collect {
      delay(1000)
      println("Data: $it")
    }
    flow1.collectLatest {  }
  }
  delay(10000)
}